package com.shujia.flink.core

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import java.sql.{Connection, DriverManager, PreparedStatement}
import java.text.SimpleDateFormat
import java.time.Duration
import java.util.Date

object Demo6Cars {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //1、读取数据
    val linesDS: DataStream[String] = env.readTextFile("flink/data/cars_sample.json")

    //2、解析数据
    val carsDS: DataStream[(Long, Long, Double)] = linesDS.map(line => {
      //将json字符串转换成json对象
      val jsonObj: JSONObject = JSON.parseObject(line)
      //通过字段名获取值
      //卡口编号
      val card: Long = jsonObj.getLong("card")
      //车辆经过卡口的时间，转换成毫秒级别
      val time: Long = jsonObj.getLong("time") * 1000
      //速度
      val speed: Double = jsonObj.getDouble("speed")
      (card, time, speed)
    })

    //3、设置时间字段和水位线
    val ws: WatermarkStrategy[(Long, Long, Double)] = WatermarkStrategy
      //水位线生成策略
      .forBoundedOutOfOrderness[(Long, Long, Double)](Duration.ofSeconds(5))
      //时间字段
      .withTimestampAssigner(new SerializableTimestampAssigner[(Long, Long, Double)] {
        override def extractTimestamp(element: (Long, Long, Double), recordTimestamp: Long): Long = element._2
      })
    val assDS: DataStream[(Long, Long, Double)] = carsDS.assignTimestampsAndWatermarks(ws)

    /**
     * 实时统计卡扣拥堵情况
     * 1、统计最近15分钟的车辆，每隔5分钟更新一次
     * 2、统计车流量和平均车速
     * 犯规的数据格式，卡扣编号，时间，车流量，平均车速
     *
     *
     */

    //按照卡扣编号分组
    val keyByDS: KeyedStream[(Long, Long, Double), Long] = assDS.keyBy(_._1)

    //划分窗口
    val windowDS: WindowedStream[(Long, Long, Double), Long, TimeWindow] = keyByDS
      .window(SlidingEventTimeWindows.of(Time.minutes(15), Time.minutes(5)))

    /**
     * apply：划分窗口之后使用apply方法，用于处理窗口的数据的方法，可以获取到一个窗口内素有的数据
     */
    ///计算车流量和平均车速
    val cardFlowAndSpeedDS: DataStream[(Long, String, Long, Double)] = windowDS
      .apply(new WindowFunction[(Long, Long, Double), (Long, String, Long, Double), Long, TimeWindow] {

        /**
         * apply:每一个key每一个窗口执行一次
         *
         * @param card   ：分组的key
         * @param window :窗口对象，可以获取到窗口的开始和结束时间
         * @param input  ：窗口内所有的数据
         * @param out    ：用于将结果发送到下游
         */
        override def apply(card: Long,
                           window: TimeWindow,
                           input: Iterable[(Long, Long, Double)],
                           out: Collector[(Long, String, Long, Double)]): Unit = {

          var flow = 0
          var sumSpeed = 0.0
          for ((_, _, speed) <- input) {
            //计算车流量
            flow += 1
            //总的车速
            sumSpeed += speed
          }
          //平均车速
          val avgSpeed: Double = sumSpeed / flow

          //获取窗口时间
          val endTime: Long = window.getEnd
          //将时间戳转换成时间字符串
          val date = new Date(endTime)
          val format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss")
          val endDate: String = format.format(date)

          //发送数据到下游
          out.collect((card, endDate, flow, avgSpeed))
        }
      })

    //将计算结果保存到mysql
    cardFlowAndSpeedDS.addSink(new RichSinkFunction[(Long, String, Long, Double)] {

      var con: Connection = _

      override def open(parameters: Configuration): Unit = {
        //加载驱动创建数据库连接
        Class.forName("com.mysql.jdbc.Driver")
        con = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
      }

      override def close(): Unit = {
        con.close()
      }

      override def invoke(value: (Long, String, Long, Double), context: SinkFunction.Context): Unit = {
        //需要先在mysql中创建表
        val stat: PreparedStatement = con.prepareStatement("insert into card_flow_avg_speed values(?,?,?,?)")
        stat.setLong(1,value._1)
        stat.setString(2,value._2)
        stat.setLong(3,value._3)
        stat.setDouble(4,value._4)

        stat.execute()
      }
    })

    env.execute()
  }

}
